package zju.dgm.service.name;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import zju.dgm.dto.ServerInfo;
import zju.dgm.service.ServiceEntry;
import zju.dgm.util.CacheManager;
import zju.dgm.util.Paths;
import zju.dgm.util.Serializer;
import zju.dgm.util.SyncUtil;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.net.InetAddress;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @author lzy
 */
@Component
public class NameService extends ServiceEntry {

    private static final Logger logger = LoggerFactory.getLogger(NameService.class);

    @Resource
    Paths paths;

    @Value("${server.port}")
    int serverPort;

    @Resource
    CuratorFramework client;

    @Resource
    CacheManager cacheManager;

    public static CountDownLatch finLatch = new CountDownLatch(1);

    @Override
    public String getRootPath() {
        return paths.getNamePath();
    }

    @Override
    protected CountDownLatch[] getPrerequisites() {
        return new CountDownLatch[0];
    }


    @Override
    protected void registerListener() {
        // 监听并写入ServerInfo,在节点挂掉时删除ServerInfo
        PathChildrenCache serverInfoCache = cacheManager.getPathChildrenCache(getRootPath());
        serverInfoCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                if (event.getData() == null) {
                    return;
                }
                String path = event.getData().getPath();
                logger.info("命名服务监听到节点变化,路径：{}", path);
                if (event.getType() == PathChildrenCacheEvent.Type.CHILD_UPDATED) {// 不管update，因为其只是用于获取id
                    byte[] nodeData = client.getData().forPath(path);
                    if (nodeData != null && nodeData.length != 0) {
                        SyncUtil.serverInfoMap.put(
                                path.substring(path.lastIndexOf('/') + 1),
                                Serializer.getObjectFromBytes(nodeData, ServerInfo.class)
                        );
                        logger.info("读取到节点信息：{}", Serializer.getObjectFromBytes(nodeData, ServerInfo.class));
                    }
                } else if (event.getType() == PathChildrenCacheEvent.Type.CHILD_REMOVED) {
                    logger.info("命名服务发现有节点被删除:{}", path.substring(path.lastIndexOf('/') + 1));
                    SyncUtil.serverInfoMap.remove(path.substring(path.lastIndexOf('/') + 1));
                }
            }
        });


    }

    @Override
    public void start() {
        logger.info("开始命名服务");
        String path = null;
        try {
            path = client.create()
                    .withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
                    .forPath(getRootPath() + "/server-",null);
        } catch (Exception e) {
            e.printStackTrace();
        }


        assert path != null;
        SyncUtil.serverName = path.substring(path.lastIndexOf("/") + 1);
        logger.info("my name is " + SyncUtil.serverName);

        try {
            ServerInfo info = new ServerInfo(SyncUtil.serverName,
                    InetAddress.getLocalHost().getHostAddress(),
                    serverPort + 1);// 简单默认在一个局域网
            client.setData().forPath(getRootPath() + "/" + SyncUtil.serverName,
                    Serializer.getBytesFromObject(info));// 写入信息
            SyncUtil.serverInfo = info;
        } catch (Exception e) {
            e.printStackTrace();
        }

        // 每分钟刷新一下Map信息
        SyncUtil.tpe.submit(() -> {
            while (true) {
                try {
                    List<String> names = client.getChildren().forPath(getRootPath());
                    for (String name : names) {
                        SyncUtil.serverInfoMap.put(
                                name,
                                Serializer.getObjectFromBytes(
                                        client.getData().forPath(getRootPath() + "/" + name),
                                        ServerInfo.class
                                )
                        );
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    try {
                        logger.info("命名服务维护信息：{}", SyncUtil.serverInfoMap);
                        finLatch.countDown();
                        Thread.sleep(60000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });

        try {
            // 等待第一次更新完毕
            finLatch.await(10, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        logger.info("完成命名服务");


    }


}

